feat(auth): [aiohttp] Add mTLS reconfiguration logic when certificate mismatch for existing credentials & Agent Identity workloads - #18224

In _recover_auth_state(), catching NotImplementedError logs a debug message without returning response, allowing execution to fall through to return None (which signals successful recovery). This causes request() to dispatch two redundant retries with the unchanged credentials before finally returning the 401 response.

Consider returning response directly inside the except NotImplementedError: block (matching the RefreshError handling below):

                                except NotImplementedError:
                                    _LOGGER.debug(
                                        "Credentials do not implement refresh()."
                                    )
                                    return response

nit: AsyncAuthorizedSession now delegates mTLS parameter checking to google.auth.aio.transport.mtls.check_parameters_for_unauthorized_response (which already handles missing certificates at lines 203–204). Because synchronous transports do not pass client_cert_callback, adding this parameter to _mtls_helper.py is unused. Consider reverting the changes to _mtls_helper.py to eliminate dead code and keep this PR focused on the aio package.

When cached_cert is unset or falsy, assigning cached_fingerprint = current_fingerprint causes cached_fingerprint != current_cert_fingerprint in sessions.py to evaluate to False. If a session is initialized with a callback that initially yields (None, None), subsequent 401 recovery on an mTLS endpoint will detect no fingerprint change once certificates become available, skipping channel reconfiguration.

Consider assigning cached_fingerprint = None when cached_cert is falsy so the mismatch check triggers reconfiguration:

        if cached_cert:
            cached_fingerprint = _agent_identity_utils.get_cached_cert_fingerprint(
                cached_cert
            )
        else:
            cached_fingerprint = None
        return cached_fingerprint, current_cert_fingerprint

nit: _recover_auth_state() only clears self._mtls_init_task when done() is true. If configure_mtls_channel() was called concurrently and is still pending, the call below returns the existing in-flight task without installing the new certificate callback, resulting in a redundant 401 retry. Consider awaiting self._mtls_init_task if it is still running before resetting it to None:

if self._mtls_init_task is not None:
    if not self._mtls_init_task.done():
        try:
            await self._mtls_init_task
        except Exception:
            pass
    self._mtls_init_task = None
await self.configure_mtls_channel(
    lambda: (call_cert_bytes, call_key_bytes)
)

Returning response at line 414 inside the certificate inspection except block exits _recover_auth_state() before reaching await self._credentials.refresh(self._auth_request) at line 470.

When a 401 is caused by token expiration or revocation rather than cert rotation, skipping refresh breaks caller retries: because before_request checks only local timestamps against self.expiry, it keeps re-attaching the rejected token on subsequent calls. A transient inspection error (such as a disk I/O glitch or callback failure) therefore leaves the session stuck failing with 401.

Note that moving self._mtls_check_counter += 1 to a finally block would advance the counter on MutualTLSChannelError, causing concurrent tasks to swallow reconfiguration failures.

To fix this, consider removing return response so execution falls through to credential refresh, while keeping self._mtls_check_counter += 1 in the else block at line 458.

nit: In configure_mtls_channel(), old transports are buffered until len(self._old_auth_requests) >= 2 before closing. Because a single configuration call appends session._auth_request without calling close(), the eviction loop and exception suppression block in sessions.py remain unexercised here (the mock error is only caught at teardown in session.close()). Consider driving configure_mtls_channel() three times (clearing _mtls_init_task = None between calls) or pre-populating _old_auth_requests so the test exercises the eviction loop and verifies that close() exceptions do not abort configuration.

In configure_mtls_channel(), self._auth_request is reassigned to the new AiohttpRequest before the eviction loop, while old_auth_request is only appended to self._old_auth_requests after the loop completes. If the task is cancelled while awaiting oldest_auth_request.close(), old_auth_request is never tracked in self._old_auth_requests, causing session.close() to skip closing its underlying session and connection pool.

Popping oldest_auth_request before awaiting its close also risks leaving the in-eviction transport unclosed if cancellation occurs during the await.

Consider appending old_auth_request immediately, updating the eviction threshold to > 2, and popping oldest_auth_request only after its close completes:

                            old_auth_request = self._auth_request
                            self._auth_request = AiohttpRequest(session=new_session)
                            self._old_auth_requests.append(old_auth_request)
                            while len(self._old_auth_requests) > 2:
                                oldest_auth_request = self._old_auth_requests[0]
                                try:
                                    if hasattr(oldest_auth_request, "close"):
                                        res = oldest_auth_request.close()
                                        if inspect.isawaitable(res):
                                            await res
                                except Exception:
                                    pass
                                self._old_auth_requests.pop(0)

nit: max_allowed_time=0.01 (10ms) against real clock time risks flaking on loaded CI runners if event loop scheduling latency exceeds ~8.5ms during initial mock dispatch, which aborts before 401 recovery is entered and leaves mock_resp_401 unclosed. Consider setting max_allowed_time=0.1 (or 0.2) to provide ample headroom for mock dispatch while keeping unit test runtime brisk and reliably timing out inside slow_refresh (10s):

        with pytest.raises(TimeoutError):
            await session.request("GET", "https://example.com", max_allowed_time=0.1)

nit: In test_cert_rotation_success_and_retry and test_psc_endpoint_triggers_cert_rotation, asserting mock_conf.assert_called_once_with(mock.ANY) accepts any argument and does not verify that the newly fetched certificates are forwarded into configure_mtls_channel(). Consider capturing the callback and asserting its return value to ensure rotated credentials are sent:

            mock_conf.assert_called_once()
            cb = (
                mock_conf.call_args.args[0]
                if mock_conf.call_args.args
                else mock_conf.call_args.kwargs["client_cert_callback"]
            )
            assert cb() == (new_cert, new_key)

nit: The docstring lists only non-None types for the 4-tuple return, but lines 203–204 return (None, None, None, None) when mTLS is inactive or certificates are absent. Consider documenting the sentinel return and using Optional types in the docstring to keep the return contract explicit:

    Returns:
        Tuple[Optional[bytes], Optional[bytes], Optional[str], Optional[str]]:
            is_mtls, current_cert_fingerprint, call_cert_bytes, call_key_bytes.
            Returns (None, None, None, None) if mTLS is disabled or no client certificate is present.